perf: parallelize serial stages around group-by (binary aggs, OP_IF, filter/where/gather)#338
perf: parallelize serial stages around group-by (binary aggs, OP_IF, filter/where/gather)#338ser-vasilich wants to merge 10 commits into
Conversation
Route binary co-moment aggregators through the dense-array (DA) group path instead of the hash scatter path. Adds sum_y/sumsq_y/sumxy co-moment slots to da_accum_t + per-row accumulation + per-worker merge; emit_agg_columns already finalises PEARSON/COV/WAVG/WSUM from the co-moments. Fixes poor multi-thread scaling of by-key binary aggregates (was ~2x, DA path scales ~9-12x like stddev). Verified vs numpy; diff comparator relaxed to 1e-9 combined abs+rel (1e-12 absolute tested bit-identical summation). Includes temporary RAY_GRPPROF phase instrumentation (to remove). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
wavg/pearson accumulate Sx as double even when the x column is integer (e.g. wavg(bsize,bid), bsize=I32). The per-worker merge dispatched on the x-column type -> read the double bits as int64 -> garbage at >1 worker. Force float merge for binary aggs at all 3 sum-merge sites. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The parallel slot-range merge (da_merge_fn, taken when n_slots>=1024) merged sumsq but not the binary-aggregate co-moment arrays (sum_y/sumsq_y/sumxy). Multi-key pearson/cov/wavg over >=1024 dense slots produced wrong results at >1 worker. Add the DA_NEED_PAIR merge. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The binary-agg DA fix lands on the default path (v2 declines CHAR-keyed binary group-bys -> legacy DA), so the debug env overrides are no longer needed. 3635/3635 tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
… eager exec_if always took the 'selected' lazy-branch path, whose scaffolding (true-count, id-list build, per-branch gather, scatter) is serial over ALL rows — every if-projection ran at single-core speed regardless of -c (100M numeric if: 2.2s at any core count). 1. exec_if_eager: one shared fixed-width elementwise fill, dispatched across the worker pool for len >= 64K (SYM sides warm their runtime-id LUT serially first — sym.c frozen-table rule, mirrors window.c). STR keeps the serial append path. 2. exec_if_selected: bail to eager when both branches are trivial (column scan / scalar const) and eager fills the type combination correctly — the lazy path only pays off when a branch is an expression worth restricting to its passing rows. Mixed numeric/string shapes stay on the selected path (its per-value string conversion). 100M rows local c24: numeric if 1883->310ms, sym if 2012->306ms. dazzle c48 canonical Q22: 2658->1333ms end-to-end. make test 3635/3635. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…pact exec_filter ran two sequential 0..nrows sweeps (pass-count and match_idx build) before its parallel gather; sel_compact rebuilt match_idx from the rowsel serially. Both now use the classic 3-phase compaction: parallel per-chunk/per-seg counts, tiny serial prefix, parallel fill at disjoint offsets. Lazy/morsel-backed predicates keep the sequential sweep. 100M rows local c24: 2-col where-select 143->21.7ms (1.4x -> 6.6x scaling); if+where 1040->324ms. make test 3635/3635. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
- ray_where_fn: 3-phase chunk compaction on the pool (was fully serial). - gather_by_idx: fixed-width value gathers dispatched over disjoint output ranges (null-bit propagation stays serial - shared-word bit writes would race). - exec_filter/where chunk phases now use ray_pool_dispatch_n (one task per chunk); ray_pool_dispatch morselizes total_elems by 1024, so passing chunk counts gave only ~2 tasks for 100M rows. 100M rows local c24: where 88->25ms, at-gather 80->31ms, 2-col where-select 138->20ms (6.8x). make test 3635/3635. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Blockers (DA binary-agg y-column): - eligibility now requires a plain numeric/temporal y; nullable integer/temporal y stays on the HT path (da_accum_row's pair branch has no y-side sentinel machinery - nulls would accumulate as values) - an FP y with HAS_NULLS sets da_any_nullable so nn[] is allocated and the emitter divides by the non-null PAIR count, not the group count Majors: - all new parallel gates require pool->n_workers > 0 (a -c 1 pool exists with 0 workers; ring fill + atomics + rc_sync were pure overhead, and the OP_IF eager reroute lost to the selected path serially - the Q22/Q25 c1 regression) - chunked dispatch_n call sites cap chunks at 1024 = the pool's initial ring capacity, so the ring never grows (dispatch_n clamps and silently DROPS tasks if ring growth fails -> uninitialized prefix entries -> OOB writes) - sel_compact seg fill switched to dispatch_n over seg-chunks (ray_pool_dispatch over segs gave 1 task under 8.4M rows) - gather_by_idx parallel path guarded by ray_parallel_flag == 0 (leaf utility, 35+ call sites; nested dispatch would corrupt the single-producer task ring) Nits: stray time.h include, restored v2-gate comment, RAY_PARALLEL_THRESHOLD symbol in pivot.c. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
|
Audit findings addressed — point by point: Blocker 1 (y-side null semantics diverge between paths). Fixed by unifying, and the fix went further than the suggested minimum. The legacy HT accumulators now implement the same pair-skip contract as the scalar reducer, the v2 engine and the DA path: a null on either side voids the whole (x,y) pair ( Chasing this exposed two more latent bugs in the same code, both fixed:
Blocker 2 (no tests). Added 3 (inconsistent nested-dispatch guard). Added 5 (dead 6 (wasted parallel count for vector inputs). The premise is not quite right: 7 (threshold inconsistency). Both the gate and the table fallback now derive from one value ( 8 (short y OOB reachable via DA). The DA eligibility gate now rejects
|
|
Superseded by #341 — the audit-gate blockers (y-side null semantics + missing regression tests) are fixed there, and the description was reworked. |
2 participants
perf: parallelize the serial stages around group-by (binary aggs, OP_IF, filter/where/gather)
Problem
On a 48-core machine the TAQ benchmark (medium NYSE dataset, quote = 296.6M rows) showed group-by queries scaling only 2.1–2.3× from 1 to 48 cores. Profiling the full query pipeline showed the group-by kernel itself was fine after per-worker accumulation — the wall-clock was dominated by serial stages around it:
pearson/cov/wavg/wsum) were not DA-eligible → fell to slower paths;OP_IFprojections went throughexec_if_selected(serial per-row id-scaffolding) even when no WHERE was present;exec_filter/sel_compactwas serial;wherebuiltin andat col idxgather were serial;ray_pool_dispatchwas being fed chunk counts, so 1500+ chunks morselized into ~2 tasks.What changed
src/ops/group.c— binary aggregates via the DA (dense-array) path: per-worker accumulators extended with co-moments (Σx, Σx², Σy, Σy², Σxy), finals computed inemit_agg_columns. Two merge bugs fixed along the way: Σx for integer x-columns is stored as double but was merged through theint64union member (garbage like 1.3e305 at high core counts); theda_merge_fnpath didn't merge the PAIR co-moment arrays at all (wrong pearson with ≥1024 slots). Eligibility is gated: y-column must be FP or non-nullable integer — nullable-int y falls back, since DA reads raw values and would sum null encodings as data.src/ops/pivot.c—OP_IF: (1) the eager elementwise fill is parallel (type-switched morsel fill; SYM LUT warm-up stays serial before dispatch; STR output stays serial); (2) trivial-branch ifs (scan/const through alias/materialize) are rerouted fromexec_if_selectedto the eager path, gated by an out-type whitelist —exec_if_selectedfires even without a WHERE clause and was the hidden per-row serial cost in the group-by queries' projections.src/ops/filter.c—exec_filterandsel_compact: bitmap→match_idxbuild parallelized with 3-phase compaction (per-chunk popcount → serial exclusive prefix sum → per-chunk fill at disjoint offsets), one task per chunk viaray_pool_dispatch_n. Chunk count capped at 1024 (the initial task-ring capacity), so ring growth can never fail and silently drop tasks.src/ops/builtins.c/src/lang/eval.c—wherebuiltin parallel with the same 3-phase pattern (keeps the word-at-a-time zero-skip in the emit phase);gather_by_idxparallel with aray_parallel_flagreentrancy guard (existing callers may already be inside a dispatch); null-bit propagation stays serial.All parallel paths share the same gates:
pool && pool->n_workers > 0(at-c 1the pool exists with 0 workers),len >= RAY_PARALLEL_THRESHOLD, serial fallback on any allocation failure.Results
48-core box, canonical
queryEngines.sh, medium TAQ, min of 3 runs, both binaries built from the same dev base (8f8033c):Side wins from the filter/where/if work: Q48 2.28×, Q42 1.69×, Q47 1.63×, Q7 1.49×, Q39 1.29×, Q8/Q16/Q17/Q19/Q41 1.2–1.3×.
Single-core: no regressions — branch ≤ dev at c1 across all 53 queries (e.g. Q22: 8052 vs 8216 ms), thanks to the
n_workers > 0gates.Correctness
make test: 3639/3639 on a clean build.wavg/pearsonresults verified against numpy to full precision at c1/c4/c24, including FP-y columns with NaNs and integer x-columns.Known remaining gaps (out of scope here)
in,if, gather, arithmetic) materializes its full result before the next starts. Fusing the filter/projection expressions into one compaction pass would remove most of the remaining wall-clock — architectural follow-up.window.cPass 3); low-cardinality partitioning (e.g. by exchange = 19 parts) caps their scaling.🤖 Generated with Claude Code
https://claude.ai/code/session_01JZwH1gpAeLzDX2bxc4jz6s